package com.dx.chat.kafka.config;

import com.dx.chat.kafka.core.KafkaConfig;
import com.dx.chat.kafka.core.KafkaParentProperties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.Map;

/**
 * kafka多消费者实例配置
 *
 * @author wcx
 * @date 2023/6/28
 */
@Configuration
public class KafkaMultiConsumerRegister implements BeanPostProcessor {

    @Resource
    private KafkaConfig kafkaConfig;

    @Resource
    private DefaultListableBeanFactory defaultListableBeanFactory;

    private boolean initializingContext = true;

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if (this.initializingContext) {
            initMultiConsumer();
        }
        return bean;
    }

    public void initMultiConsumer() {
        this.initializingContext = false;
        if (CollectionUtils.isEmpty(kafkaConfig.getDxTopics())) {
            return;
        }
        kafkaConfig.getDxTopics().forEach(dxTopic -> {
            dxTopic.getConsumers().forEach(consumer -> {
                // 获取消费者配置
                Map<String, Object> commonProperties = buildConsumerFactoryProperties(kafkaConfig, consumer);
                // 注入ioc
                ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
                // 设置并发量，小于或者等于 Topic 的分区数
                factory.setConcurrency(dxTopic.getNumPartitions());
                // 设置提交偏移量的方式 手动
                if (Boolean.TRUE.equals(!consumer.getEnableAutoCommit())) {
                    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
                }
                // 设置批量消费
                if (Boolean.TRUE.equals(consumer.getBulkConsumption())) {
                    factory.setBatchListener(Boolean.TRUE);
                }
                factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(commonProperties));
                defaultListableBeanFactory.registerSingleton(consumer.getConsumerBeanName(), factory);
            });
        });

    }

    /**
     * 建立消费者工厂属性
     *
     * @param kafkaConfig
     * @param consumer
     * @return {@link Map}<{@link String}, {@link Object}>
     */
    private Map<String, Object> buildConsumerFactoryProperties(KafkaConfig kafkaConfig, KafkaParentProperties.Consumer consumer) {

        Map<String, Object> commonProperties = kafkaConfig.buildCommonProperties();
        commonProperties.putAll(consumer.buildProperties());
        // 接受反序列化任意的类，也可限定包路径，限定包路径在公共配置中
        if (!commonProperties.containsKey(JsonDeserializer.TRUSTED_PACKAGES) &&
                commonProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG).equals(JsonDeserializer.class)) {
            commonProperties.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        }
        // 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
        commonProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");

        // 批量消费
        if (Boolean.TRUE.equals(consumer.getBulkConsumption())) {
            // 两次Poll之间的最大允许间隔。
            // 消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。
            commonProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
            // 设置单次拉取的量，走公网访问时，该参数会有较大影响。
            commonProperties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
            commonProperties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
        }

        return commonProperties;
    }

}
